package top.lywivan.netty.reactor.cnxn;


import top.lywivan.netty.reactor.Server;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

public class NIOServerCnxn extends ServerCnxn{

    private final SocketChannel sc;
    private final SelectionKey key;
    private final NIOServerCnxnFactory factory;
    private final NIOServerCnxnFactory.SelectorThread selectorThread;
    private final InetSocketAddress addr;

    private final boolean clientTcpKeepAlive = Boolean.getBoolean("clientTcpKeepAlive");

    // tcp
    private final ByteBuffer headerBuffer =  ByteBuffer.allocate(4);
    protected ByteBuffer incomingBuffer = headerBuffer;
    private final Queue<ByteBuffer> outgoingBuffers = new LinkedBlockingQueue<ByteBuffer>();


    //是否已建立会话
    private boolean initialized;


    public NIOServerCnxn(Server server, SocketChannel sc, SelectionKey key, NIOServerCnxnFactory factory, NIOServerCnxnFactory.SelectorThread selectorThread) throws IOException {
        super(server);
        this.sc = sc;
        this.key = key;
        this.factory = factory;
        this.selectorThread = selectorThread;

        //
        sc.socket().setTcpNoDelay(true);
        sc.socket().setSoLinger(false,-1);
        sc.socket().setKeepAlive(clientTcpKeepAlive);
        addr = (InetSocketAddress)sc.socket().getRemoteSocketAddress();

    }

    public void close(DisconnectReason serverShutdown) {
        // cancel key
        if (key != null) {
            try {
                // need to cancel this selection key from the selector
                key.cancel();
            } catch (Exception e) {
            }
        }
        // close socket
        if (!sc.isOpen()) {
            return;
        }
        try {
            sc.socket().shutdownOutput();
        } catch (IOException e) {
            e.printStackTrace();
        }
        try {
            sc.socket().shutdownInput();
        } catch (IOException e) {
        }
        try {
            sc.socket().close();
        } catch (IOException e) {

        }
        try {
            sc.close();
        } catch (IOException e) {
        }
    }

    public static void closeSock(SocketChannel sock) {
        if (!sock.isOpen()) {
            return;
        }

        try {
            /*
             * The following sequence of code is stupid! You would think that
             * only sock.close() is needed, but alas, it doesn't work that way.
             * If you just do sock.close() there are cases where the socket
             * doesn't actually close...
             */
            sock.socket().shutdownOutput();
        } catch (IOException e) {
            // This is a relatively common exception that we can't avoid
        }
        try {
            sock.socket().shutdownInput();
        } catch (IOException e) {
            // This is a relatively common exception that we can't avoid
        }
        try {
            sock.socket().close();
        } catch (IOException e) {
        }
        try {
            sock.close();
        } catch (IOException e) {
        }
    }


    /**
     * Handles read/write IO on connection.
     */
    public void doIO(SelectionKey key) {
        try {
            if (!isSocketOpen()) {
                return;
            }

            //处理读事件
            if (key.isReadable()) {
                int i = sc.read(incomingBuffer);
                if (i < 0) {
                    handleFailedRead();
                }

                // 等到 incomingBuffer 读满
                if (incomingBuffer.remaining() == 0) {

                    boolean isBody = false;

                    // 如果header 读满
                    if (incomingBuffer == headerBuffer) {
                        incomingBuffer.flip();
                        // 读 body 的长度
                        isBody = readHeader(key);
                        incomingBuffer.clear();
                    }else {
                        isBody = true;
                    }

                    // 如果 body 读满
                    if (isBody) {
                        readBody();
                    }else {
                        return;
                    }

                }
            }

            //处理可写事件
            if (key.isWritable()) {
                handleWrite(key);
            }
        } catch (EndOfStreamException e) {
            e.printStackTrace();

            close(e.getReason());

        }catch (IOException e) {
            e.printStackTrace();

            close(DisconnectReason.IO_EXCEPTION);
        }

    }

    // 处理写
    private void handleWrite(SelectionKey key) throws IOException {
        if (outgoingBuffers.isEmpty()) {
            return;
        }
        ByteBuffer[] buffers = new ByteBuffer[outgoingBuffers.size()];
        sc.write(outgoingBuffers.toArray(buffers));

        // 移除已发送完成的buffer
        ByteBuffer bb;
        while ((bb = outgoingBuffers.peek()) !=null) {
            if (bb.remaining() > 0) {
                break;
            }
            outgoingBuffers.remove();
        }
    }

    private void readBody() throws IOException {
        if (incomingBuffer.remaining() !=0) {
            int rc = sc.read(incomingBuffer); // sock is non-blocking, so ok
            if (rc < 0) {
                handleFailedRead();
            }
        }

        // 处理 body 数据
        if (incomingBuffer.remaining() ==0) {
            incomingBuffer.flip();
            // TODO 应用层面是否需要先建立会话然后操作
            /*if (!initialized) {
                // 先建立会话
                doSessionRequest();
            }else {
                // 读取业务数据
                doRequest();
            }*/

            doRequest();

            headerBuffer.clear();
            incomingBuffer = headerBuffer;
        }
    }

    // 数据已读取完成 ,序列化操作后 可执行业务
    private void doRequest() {
        server.processPacket(this,incomingBuffer);
    }

    private void doSessionRequest() {
        // TODO 还未设计
    }

    private boolean readHeader(SelectionKey key) throws IOException{
        int len = headerBuffer.getInt();
        if (len < 0 || len > maxBuffer) {
            throw new IOException("Len error. "
                    + "A message from " +  this.getRemoteSocketAddress() + " with advertised length of " + len
                    + " is either a malformed message or too large to process"
                    + " (length is greater than maxbuffer=" + maxBuffer + ")");
        }
        // 根据 header的值分配要读取的body的大小
        incomingBuffer = ByteBuffer.allocate(len);
        headerBuffer.clear();
        return true;
    }

    private void handleFailedRead() throws EndOfStreamException {
        throw new EndOfStreamException("Unable to read additional data from client,"
                + " it probably closed the socket:"
                + " address = " + sc.socket().getRemoteSocketAddress(),
                DisconnectReason.UNABLE_TO_READ_FROM_CLIENT);
    }


    protected boolean isSocketOpen() {
        return sc.isOpen();
    }

    public InetSocketAddress getRemoteSocketAddress() {
        if (!sc.isOpen()) {
            return null;
        }
        return (InetSocketAddress) sc.socket().getRemoteSocketAddress();
    }


    /**
     * 发送 buffer
     * @param byteBuffers
     */
    public void sendBuffer(ByteBuffer ... byteBuffers) {
        synchronized (outgoingBuffers) {
            for (ByteBuffer byteBuffer : byteBuffers) {
                outgoingBuffers.add(byteBuffer);
            }
        }
        // 将key添加到 update queue, 让selector 继续 检测
        requestInterestOpsUpdate();
    }


    private boolean getReadInterest() {
        // return !throttled.get();
        return true;
    }

    // 是否要针对当前连接进行 限流 (停止读取数据 )
    private final AtomicBoolean throttled = new AtomicBoolean(false);

    // Throttle acceptance of new requests. If this entailed a state change,
    // register an interest op update request with the selector.
    //
    // Don't support wait disable receive in NIO, ignore the parameter
    public void disableRecv(boolean waitDisableRecv) {
        if (throttled.compareAndSet(false, true)) {
            requestInterestOpsUpdate();
        }
    }

    // Disable throttling and resume acceptance of new requests. If this
    // entailed a state change, register an interest op update request with
    // the selector.
    public void enableRecv() {
        if (throttled.compareAndSet(true, false)) {
            requestInterestOpsUpdate();
        }
    }

    public void requestInterestOpsUpdate() {
        if (isSelectable()) {
            selectorThread.addInterestOpsUpdateRequest(key);
        }
    }

    @Override
    public int getInterestOps() {
        if (!isSelectable()) {
            return 0;
        }
        int interestOps = 0;

        // 在不需要针对当前连接限流的情况下可以继续 检测其 read 事件
        if (getReadInterest()) {
            interestOps |= SelectionKey.OP_READ;
        }

        // 如果有需要向外写出的数据 ,添加 write 事件
        if (getWriteInterest()) {
            interestOps |= SelectionKey.OP_WRITE;
        }
        return interestOps;
    }


    // 标记当前连接 key 是否可以继续 检测
    private final AtomicBoolean selectable = new AtomicBoolean(true);

    public boolean isSelectable() {
        return key.isValid() && selectable.get();
    }

    public void disableSelectable() {
        selectable.set(false);
    }

    public void enableSelectable() {
        selectable.set(true);
    }


    private boolean getWriteInterest() {
        return !outgoingBuffers.isEmpty();
    }
}
